package com.shujia.streaming

import org.apache.spark.sql.SparkSession
import org.apache.spark.streaming.{Durations, StreamingContext}
import org.apache.spark.streaming.dstream.DStream

object Demo02UpdateStateByKey {
  def main(args: Array[String]): Unit = {

    // 构建SparkContext
    val spark: SparkSession = SparkSession
      .builder()
      .appName("Demo02UpdateStateByKey")
      .master("local[2]")
      .getOrCreate()

    /**
     * 创建SparkStreaming的入口
     * 需要SparkContext以及处理的间隔时间
     * 每5s处理一次---> 每5s将接收到的数据 封装成 一个RDD进行处理
     */
    val ssc: StreamingContext = new StreamingContext(spark.sparkContext, Durations.seconds(5))

    // 使用有状态算子需要指定checkpoint目录用于保存状态
    ssc.checkpoint("Spark/data/checkpoint")

    /**
     * yum install nc
     * nc -lk 8888
     */

    // 可以通过socket连接nc命令创建的服务器 模拟消息队列
    val lineDS: DStream[String] = ssc.socketTextStream("master", 8888)

    val wordsDS: DStream[String] = lineDS
      .flatMap(line => line.split(","))

    val wordsKVDS: DStream[(String, Int)] = wordsDS.map(word => (word, 1))

    /**
     * reduceByKey 只能统计当前批次内的单词数量 无法结合上次计算结果
     */
    wordsKVDS
      .reduceByKey((x: Int, y: Int) => x + y)
    //      .print()

    /**
     * updateStateByKey 有状态算子
     * 可以将当前批次计算的结果 同上一个批次的计算结果加起来 得到一个汇总后的结果
     * 需要接收一个函数f
     * 函数f的类型为：(Seq[V], Option[S]) => Option[S]
     * 输入两个参数，类型分别为Seq、Option
     * 返回值类型为 Option
     * seq 表示某一个key在当前批次内的所有数据
     * opt 表示该key上一次的计算结果
     */

    def updateKey(seq: Seq[Int], opt: Option[Int]): Option[Int] = {
      // 1、统计某一个key在当前批次内的一个结果
      val sum: Int = seq.sum
      // 2、获取上一个批次计算出的该Key的结果 与当前批次计算的结果相加
      Some(opt.getOrElse(0) + sum)
    }


    wordsKVDS
      .updateStateByKey(updateKey)
      .print()


    // 启动任务
    ssc.start()
    ssc.awaitTermination()
    ssc.stop()

    /**
     * 思考：
     * 1、热门微博：每小时基于最近24小时的搜索及浏览记录统计热门微博
     * 简化：每5s统计最近15s内的单词数量 ===> 滑动窗口
     * 2、缉查布控===>抓犯人
     * 黑名单：MySQL
     * 车辆或人员信息记录
     *
     *
     */

  }

}
